package com.yifeng.repo.toolkit.mq.configure;

import com.google.common.base.Strings;
import com.yifeng.repo.toolkit.mq.DefaultMessageProducer;
import com.yifeng.repo.toolkit.mq.OrderlyMessageQueueSelector;
import com.yifeng.repo.toolkit.mq.TempMessageConsumer;
import com.yifeng.repo.toolkit.mq.TempTransactionListener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.boot.context.properties.bind.Bindable;
import org.springframework.boot.context.properties.bind.Binder;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.context.EnvironmentAware;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;
import org.springframework.messaging.Message;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
import java.util.function.Consumer;

/**
 * Created by daibing on 2022/9/25.
 */
@Slf4j
@Configuration
@EnableConfigurationProperties(MqProperties.class)
@ConditionalOnProperty(prefix = "yfcloud.toolkit.mq", name = "enabled")
public class MqConfiguration implements EnvironmentAware {
    private final MqProperties properties;
    private List<BindingProperties> producerBindingPropertiesList;

    @Resource
    private StreamBridge streamBridge;

    public MqConfiguration(MqProperties properties) {
        this.properties = properties;
    }

    @Override
    public void setEnvironment(Environment environment) {
        Map<String, Object> bindingsMap = Binder.get(environment).bind("spring.cloud.stream.bindings", Bindable.mapOf(String.class, Object.class)).get();
        Map<String, Object> bindersMap = null;
        String defaultBinder = null;
        try {
            bindersMap = Binder.get(environment).bind("spring.cloud.stream.binders", Bindable.mapOf(String.class, Object.class)).get();
            defaultBinder = Binder.get(environment).bind("spring.cloud.stream.default-binder", String.class).get();
        } catch (NoSuchElementException e) {
            // rocket mq 不需要配置上面两个配置项
        }
        this.producerBindingPropertiesList = parseProducerBindingPropertiesList(bindingsMap, bindersMap, defaultBinder);
    }

    @Bean
    public DefaultMessageProducer defaultMessageProducer() {
        return new DefaultMessageProducer(properties, producerBindingPropertiesList, streamBridge);
    }

    @Bean
    @ConditionalOnProperty(prefix = "yfcloud.toolkit.mq", name = "enable-temp-consumer")
    public OrderlyMessageQueueSelector orderlyMessageQueueSelector() {
        return new OrderlyMessageQueueSelector();
    }

    @Bean
    @ConditionalOnProperty(prefix = "yfcloud.toolkit.mq", name = "enable-temp-consumer")
    public Consumer<Message<String>> tempMessageConsumer() {
        return new TempMessageConsumer(properties);
    }

    @Bean
    @ConditionalOnProperty(prefix = "yfcloud.toolkit.mq", name = "enable-temp-consumer")
    public TempTransactionListener tempTransactionListener() {
        return new TempTransactionListener();
    }

    @SuppressWarnings("unchecked")
    private List<BindingProperties> parseProducerBindingPropertiesList(Map<String, Object> bindingsMap, Map<String, Object> bindersMap, String defaultBinder) {
        List<BindingProperties> producerBindingPropertiesList = new ArrayList<>();
        for (Map.Entry<String, Object> entry : bindingsMap.entrySet()) {
            if (!entry.getKey().endsWith(properties.getProducerBindingSuffix())) {
                continue;
            }
            Map<String, String> bindingObj = (Map<String, String>) entry.getValue();
            BindingProperties bindingProperties = new BindingProperties();
            bindingProperties.setBindingName(entry.getKey());
            bindingProperties.setDestination(bindingObj.get("destination"));
            bindingProperties.setContentType(bindingObj.get("content-type"));
            bindingProperties.setBinder(bindingObj.get("binder"));
            if (Strings.isNullOrEmpty(bindingProperties.getBinder())) {
                bindingProperties.setBinder(defaultBinder);
            }
            if (bindersMap == null || bindersMap.isEmpty()) {
                producerBindingPropertiesList.add(bindingProperties);
                continue;
            }
            Map<String, Object> binderObj = (Map<String, Object>) bindersMap.get(bindingProperties.getBinder());
            if (binderObj == null || binderObj.isEmpty()) {
                producerBindingPropertiesList.add(bindingProperties);
                continue;
            }
            bindingProperties.setBinderType(String.valueOf(binderObj.get("type")));
            producerBindingPropertiesList.add(bindingProperties);
        }
        return producerBindingPropertiesList;
    }

}
